package com.dys.flink.wc.netty.handler;

import com.dys.flink.wc.model.Product;
import com.dys.flink.wc.util.JsonUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class ProductDecoder extends ByteToMessageDecoder {
    private static final Logger logger = LoggerFactory.getLogger(ProductDecoder.class);

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 确保有足够的数据读取长度前缀
        if (in.readableBytes() < 4) {
            return;
        }
        
        in.markReaderIndex();
        int length = in.readInt();
        
        // 确保有完整的数据包
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        
        byte[] bytes = new byte[length];
        in.readBytes(bytes);
        String json = new String(bytes, StandardCharsets.UTF_8);
        
        try {
            // 使用 Jackson 解析 JSON
            Product product = JsonUtils.fromJson(json, Product.class);
            out.add(product);
            logger.debug("Successfully decoded product: {}", product);
        } catch (Exception e) {
            logger.error("Unexpected error while decoding product: {}", json, e);
        }
    }
}